import base64
import hashlib
import hmac
import json
import math
import os
import re
import threading
import time
import traceback
import urllib
from collections import OrderedDict
from concurrent.futures import ThreadPoolExecutor, TimeoutError, as_completed
from datetime import datetime
from typing import Union

import jieba
import numpy as np
import pandas as pd
import requests
from aliyun.log import LogClient, PutLogsRequest, LogItem, GetLogsRequest, IndexConfig
import socket
import one_runtime
import random
import string
from rouge import Rouge
from nltk.translate.bleu_score import sentence_bleu, SmoothingFunction


import json

from elasticsearch import Elasticsearch
from typing import List, Dict
from urllib.parse import urlparse

from datetime import datetime
import pytz

# 链路thread_local
link_thread_local = threading.local()


def get_ip():
    try:
        host_name = socket.gethostname()
        host_ip = socket.gethostbyname(host_name)
        return host_ip
    except:
        print("Unable to get Hostname and IP")
        return None


# 日志线程局部变量
log_local = threading.local()

# 本示例从环境变量中获取AccessKey ID和AccessKey Secret。
sls_client = None
sls_project_name = None
sls_logstore_name = None
# 有配置且是服务器时才写sls
if one_runtime.get_config('SLS_ACCESS_KEY_ID') and get_ip() == one_runtime.get_config('serverip'):
    accessKeyId = one_runtime.get_config('SLS_ACCESS_KEY_ID')
    accessKey = one_runtime.get_config('SLS_ACCESS_KEY')
    # 日志服务的服务接入点。此处以杭州为例，其它地域请根据实际情况填写。
    endpoint = one_runtime.get_config('SLS_ENDPOINT')
    sls_project_name = one_runtime.get_config('SLS_PROJECT_NAME')
    sls_logstore_name = one_runtime.get_config('SLS_LOGSTORE_NAME')
    if not sls_project_name or not sls_logstore_name:
        raise ValueError('sls_config is invalid, please set sys config SLS_PROJECT_NAME,SLS_LOGSTORE_NAME')
    # 创建日志服务Client。
    sls_client = LogClient(endpoint, accessKeyId, accessKey)


# 链路记录器
class LinkInfo:
    '''
    标准用发
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 创建新的链路节点
        result = None
        now_link = LinkInfo(action, getattr(link_thread_local, 'current_link', None), args, kwargs)
        try:
            now_link.step_in()
            # 执行函数，并记录出参
            result = func(*args, **kwargs)
        except Exception as e:
            now_link.step_out(error=e)
        finally:
            now_link.step_out(result=result)
        return result
    '''

    def __init__(self, biz: str = None, action: str = None, ext: dict = None):
        self.parent = getattr(link_thread_local, 'current_link', None)
        self.group = 'one_step'
        self.biz = biz if biz else 'one_step'
        self.action = action if action else ''
        self.root_time = self.parent.root_time if self.parent else int(time.time() * 1000)
        self.begin = 0 if self.parent is None else int((time.time() * 1000 - self.root_time))
        self.end = -1
        self.segments = OrderedDict()
        self.id = self.generate_id()
        self.ext = {}
        # 入参尽量转json
        if isinstance(ext, dict):
            for key, val in ext.items():
                try:
                    v1 = json.loads(json.dumps(val, ensure_ascii=False, default=str))
                    try:
                        self.ext[key] = json.loads(v1)
                    except:
                        self.ext[key] = v1
                except:
                    self.ext[key] = str(val)
        else:
            try:
                v = json.loads(json.dumps(ext, ensure_ascii=False, default=str))
                try:
                    self.ext['req'] = json.loads(v)
                except:
                    self.ext['req'] = v
            except:
                self.ext['req'] = str(ext)

        self.status = None
        self.error = None
        self.performance = 'NORMAL'
        if self.parent:
            self.parent.segments[self.id] = self

    def generate_id(self):
        if self.parent:
            # 如果有父节点，id为父节点的id加上父节点的子节点数量
            return self.parent.id + "." + str(len(self.parent.segments) + 1)
        else:
            return "0"

    def step_in(self):
        try:
            # 设置当前链路为新的链路节点
            link_thread_local.current_link = self
        except Exception as e:
            log('step_in_error', {
                'error': e
            })

    def is_root(self):
        return self.parent is None

    def is_done(self):
        return self.end != -1

    def digest_log(self, msg: str = None):
        log('s_digest', {
            'biz': self.biz,
            'action': self.action,
            'begin': self.begin,
            'end': self.end,
            'id': self.id,
            'status': self.status,
            'error': self.error,
            'performance': self.performance, **self.ext
        }, msg)

    def link_log(self, msg: str = None):
        log('s_link', {
            'biz': self.biz,
            'action': self.action,
            'id': self.id,
            'cost': self.cost,
            'status': self.status,
            'error': self.error,
            'performance': self.performance,
            'link': json.dumps(self.to_dict(), ensure_ascii=False, default=str)
        }, msg)

    def step_out(self, result: dict = None, error=None, status: str = None):
        # 追踪结束，记录结束时间和耗时
        try:
            if self.end != -1:
                return
            self.end = int((time.time() * 1000 - self.root_time))
            self.cost = self.end - self.begin
            if result is not None:
                self.ext['result'] = json.loads(json.dumps(result, ensure_ascii=False, default=str))
                try:
                    self.ext['result'] = json.loads(self.ext['result'])
                except:
                    pass
            if error:
                if isinstance(error, dict):
                    self.status = 'unknown_error'
                    self.error = error
                else:
                    self.status = 'error_' + str(type(error))
                    self.error = {
                        'errorCode': str(type(error)),
                        'errorMsg': str(error),
                        'stacks': error_stack(error).split('\n')
                    }
            else:
                self.status = 'OK' if result is not None else "NULL"
            if status:
                self.status = status
            if self.parent:
                link_thread_local.current_link = self.parent
            else:
                del link_thread_local.current_link
        except Exception as e:
            log('step_out_error', {
                'error': e
            })

    def to_dict(self):
        return {
            'group': self.group,
            'biz': self.biz,
            'action': self.action,
            'id': self.id,
            'begin': self.begin,
            'end': getattr(self, 'end', None),  # 在还没有结束追踪时，end 属性可能不存在
            'cost': getattr(self, 'cost', None),  # 在还没有结束追踪时，cost 属性可能不存在
            'segments': {k: v.to_dict() for k, v in self.segments.items()},
            'ext': self.ext,
            'status': self.status,
            'error': self.error,
            'performance': self.performance
        }

    def __repr__(self):
        return json.dumps(self.to_dict(), ensure_ascii=False, default=str)


def log(code: str = None, map: dict = None, format_string: str = None, *args):
    '''
    输出日志
    :param format_string: 信息格式
    :param map: k-v载荷信息
    :param args: 信息填充
    :return:
    '''
    # 获取毫秒级时间戳并截断为毫秒
    current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3]
    content = format_msg(format_string, args)
    content_ = {}
    try:
        log_data: dict = getattr(log_local, "run_data", None)
        # 日志上下文
        if log_data:
            content_.update(log_data)
        if content:
            content_ = {
                'content': content
            }
        if isinstance(map, dict):
            content_.update(map)

        # 非服务器不写sls
        if not sls_client:
            print('[' + current_time + ']' + (str(code) if code is not None else '') + ' ' + str(content_))
            return
        sls(code, content_)
    except Exception as e:
        print('[' + current_time + ']' + 'slsError ' + error_stack(e))
        print('[' + current_time + ']' + (str(code) if code is not None else '') + ' ' + str(content_))


def sls(code: str = None, info: dict = None):
    if not sls_client:
        raise ValueError('sls_client is not init, please set sys config SLS_ACCESS_KEY,SLS_ACCESS_KEY_ID,SLS_ENDPOINT')
    '''
    sls日志
    :param code:
    :param info:
    :return:
    '''
    log_item = LogItem()
    contents = [('scene', str(code) if code is not None else '')]
    # 添加info的信息
    if isinstance(info, dict):
        for key, value in info.items():
            contents.append((key, desc_obj(value)))  # 将值转换为字符串
    log_item.set_contents(contents)
    request = PutLogsRequest(sls_project_name, sls_logstore_name, 'one_step', "", [log_item], compress=False)
    sls_client.put_logs(request)


def error_stack(error: Exception = None):
    if not isinstance(error, Exception):
        return ''
    tb_str = traceback.format_exception(etype=type(error), value=error, tb=error.__traceback__)
    return "".join(tb_str)


# 标准大模型输入格式
class LlmMessage():
    def __init__(self, role: str = None, content: str = None):
        '''
        :param role: user assistant system
        :param content:
        '''
        self.role = role
        self.content = content

    @classmethod
    def from_dict(cls, data: dict):
        return cls(role=data.get('role'), content=data.get('content'))

    def to_dict(self):
        return {
            "role": self.role,
            'content': self.content
        }

    # 添加 to_json 方法
    def to_json(self):
        return json.dumps(self.to_dict(), ensure_ascii=False, default=str)

    # 实现 __repr__ 方法
    def __repr__(self):
        return self.to_json()


# build_prompt替代版
def build_messages(msg: Union[str, dict, list[LlmMessage], list[dict]], obj) -> list[dict]:
    '''
    构建大模型请求的messages
    :param msg: 信息  可为单字符串 或为 LlmMessage 数组
    :param obj: 占位符值
    :return:
    '''

    def update_content(messages_obj):
        for i in range(len(messages_obj)):
            item = messages_obj[i]
            if not item:
                continue
            if not isinstance(item, dict):
                item = item.to_dict()
                messages_obj[i] = item

            content_ = item.get('content')
            if content_:
                item['content'] = build_prompt(content_, obj)
        return messages_obj

    if isinstance(msg, str):
        try:
            msg = json.loads(msg)
        except:
            msg = LlmMessage(role='user', content=msg)
    if isinstance(msg, list):
        return update_content(msg)
    if not isinstance(msg, dict):
        msg = msg.to_dict()
    content_ = msg.get('content')
    if content_:
        msg['content'] = build_prompt(content_, obj)
    return [msg]


def build_prompt(template: str, obj) -> str:
    '''
    :param template: 模板
    :param obj: 占位符的值
    :return:
    '''
    PLACEHOLDER_PATTERN = re.compile(r'\{([a-zA-Z0-9_]+)(\.([a-zA-Z0-9_ ,一-龥]+))?\}')
    if template is None:
        return None
    if obj is None:
        jsonObj = {}
    elif isinstance(obj, dict):
        jsonObj = obj
    elif isinstance(obj, str):
        jsonObj = json.loads(obj)
    else:
        jsonObj = json.loads(str(obj))
    matcher = PLACEHOLDER_PATTERN.finditer(template)
    output = ''
    i = 0
    for match in matcher:
        first_key = match.group(1)
        second_key = match.group(3)
        first_val = jsonObj.get(first_key)

        if second_key is None:
            val = first_val
        else:
            if isinstance(first_val, str):
                first_val = first_val.strip()
                if first_val == 'null' or first_val == 'None':
                    val = None
                elif first_val:
                    try:
                        # 尝试将字符串解析为 JSON 对象
                        try:
                            first_val = json.loads(first_val)
                        except:
                            # 使用正则表达式移除注释
                            try:
                                first_val = json.loads(re.sub(r'//.*', '', first_val))
                            except Exception as e:
                                log('placeholder_second_error', {
                                    'first_val': first_val,
                                    'error': e
                                })
                        val = first_val.get(second_key)
                    except Exception:
                        raise ValueError('二级参数取值失败，传入非标准json，firstKey:' + first_key + ', secondKey:' + second_key)
                else:
                    val = None
            elif isinstance(first_val, dict):
                val = first_val.get(second_key)
            else:
                val = None
        replacement = __desc_prompt_val(val)
        output += template[i:match.start()] + replacement
        i = match.end()
    output += template[i:]
    return output


def parse_llm_tools(tools: list[dict] = None):
    tools_format = """{name}: {description}，parametersSchema: {parameters}"""
    tool_desc = []
    tool_name = []
    if tools:
        for func_info in tools:
            tool_desc.append(tools_format.format(name=func_info.get('name'), description=func_info.get('description'), parameters=json.dumps(func_info.get('parameters'), ensure_ascii=False)))
            tool_name.append(func_info.get('name'))
    return (tool_name, tool_desc)


def merge_react_messages(messages: list[dict], params: dict, tools_name: list[str], tools_desc: list[str], observation: str,support_assistant_predict:bool=False):
    if not messages:
        raise ValueError('messages不可为空')
    final_sys_template = '''{sys_pre}
Answer the following questions as best you can. You have access to the following APIs:

{tool_desc}

Take a deep breath and work on this problem step-by-step,Be sure to follow the following format for your output.
Use the following format:
{sys_react_pre}
Question: the input question you must answer
Thought: you should always think about what to do
Action: when you need invoke tool, the action to take, must be one of [{tool_name}]
Action Input: the input to the action
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can be repeated zero or more times)
Thought: I now know the answer or need reply user
Reply User: the answer to the question or reply info to user
{sys_react_after}
'''
    final_sys = build_prompt(final_sys_template, {
        'sys_pre': params.get('sys_pre') if params else '',
        'tool_desc': '\n'.join(tools_desc),
        'tool_name': ','.join(tools_name),
        'sys_react_pre': params.get('sys_react_pre') if params else '',
        'sys_react_after': params.get('sys_react_after') if params else ''
    })
    messages.insert(0, {
        'role': 'system',
        'content': final_sys
    })
    # 第一句
    if messages[len(messages) - 1]['role'] == 'user':
        messages[len(messages) - 1]['content'] = 'Take a deep breath and work on this problem step-by-step,Be sure to follow the following format for your output.\nUse the following format:\n' + (params.get('user_react_pre') if params.get('user_react_pre') else '') + '\nQuestion: ' + messages[len(messages) - 1]['content']
    elif len(messages) > 2 and messages[len(messages) - 1]['role'] == 'assistant':
        # assistant添加observation
        if observation is None:
            raise ValueError('observation 不可为空')
        if support_assistant_predict:
            # 已经进度续写了，则改写前一个user
            messages[len(messages) - 2]['content'] = 'Use the following format:\n' + (params.get('user_react_pre') if params.get('user_react_pre') else '') + '\nQuestion: ' + messages[len(messages) - 2]['content']

            messages[len(messages) - 1]['content'] = messages[len(messages) - 1]['content'] + '\nObservation: ' + str(observation)
        else:
            # 最后两条数据合一
            one = 'Use the following format:\n' + (params.get('user_react_pre') if params.get('user_react_pre') else '') + '\nQuestion: ' + messages[len(messages) - 2]['content']
            two = messages[len(messages) - 1]['content'] + '\nObservation: ' + str(observation)
            messages[len(messages) - 2]['content'] = one + '\n' + two
            messages.pop()

    else:
        raise ValueError('不支持react格式，' + json.dumps(messages, ensure_ascii=False))





# 结果解析器
def parse_react_result(ori_result: str = None, finished: bool = None, tools_name: str = None) -> (str, dict, str):
    '''
    :param ori_result: 原始内容
    :param finished: 是否已经结束了
    :param tools_name: 可选工具列表
    :return: thought，工具名，工具参数，reply_user
    '''
    t_l = len("Thought:")
    t = ori_result.rfind("Thought:")
    i_l = len("\nAction:")
    i = ori_result.rfind("\nAction:")
    if i == -1:
        i = ori_result.rfind("Action:")
        i_l = i_l - 1

    j_l = len("\nAction Input:")
    j = ori_result.rfind("\nAction Input:")
    if j == -1:
        j = ori_result.rfind("Action Input:")
        j_l = j_l - 1
    k = ori_result.rfind("\nObservation:")
    if k == -1:
        k = ori_result.rfind("Observation:")
    z_l = len("\nReply User:")
    z = ori_result.rfind("\nReply User:")
    if z == -1:
        z = ori_result.rfind("Reply User:")
        z_l = z_l - 1
    #     有thought
    thought = None
    if t > -1:
        if t > -1:
            # 有Action
            if i > t:
                thought = ori_result[t + t_l: i]
            elif z > t:
                thought = ori_result[t + t_l: z]
            else:
                # Observation:和 Reply User的未完全展示的长度
                thought = ori_result[t + t_l:-13]
        else:
            # 有Action
            if i > -1:
                thought = ori_result[: i]
            elif z > -1:
                thought = ori_result[: z]
            else:
                # Observation:和 Reply User的未完全展示的长度
                thought = ori_result[:-13]

    # 工具调用¬
    if 0 <= i < j:  # If the test has `Action` and `Action input`,
        func_name = ori_result[i + i_l: j].strip()
        if func_name not in tools_name:
            raise ValueError('不存在的tool，' + func_name)
        # 如果已经结束了
        if finished:
            if k < j:  # but does not contain `Observation`,
                # then it is likely that `Observation` is omitted by the LLM,
                # because the output test may have discarded the stop word.
                ori_result = ori_result.rstrip() + "\nObservation:"  # Add it back.
            k = ori_result.rfind("Observation:")
            func_args = json.loads(ori_result[j + j_l: k].strip())
            if func_name:
                return (thought, func_name, func_args, None)
        else:
            # 没结束就不给工具参数
            return (thought, func_name, None, None)

    # 回复
    if z >= 0:
        return (thought, None, None, ori_result[z + z_l:])

    # 没有thought 当做回复看
    if t < 0 and i < 0 and len(ori_result) > 15:
        return (thought, None, None, ori_result)
    return (thought, None, None, None) if not finished else (None, None, None, ori_result)


def __desc_prompt_val(val):
    if val is None or 'nan' == val or 'None' == val or 'null' == val:
        return ''
    if isinstance(val, (float, int)) and math.isnan(val):
        return ''
    if isinstance(val, bool):
        return 'true' if val else 'false'
    elif isinstance(val, (list, tuple, set, dict)):
        return json.dumps(val, ensure_ascii=False, default=str)
    return str(val)


def split_by_size(datas: list, queue_size: int) -> list[list]:
    '''
    # 将内容根据列表长度拆分
    :param datas:
    :param queue_size:
    :return:
    '''
    if not datas or not queue_size:
        return []
    sublists = [[] for _ in range(queue_size)]
    for i, item in enumerate(datas):
        sublists[i % queue_size].append(item)
    return sublists


def run_tasks_in_parallel(task_map: dict = None, timeout=None, callback=None):
    """
    使用线程池并行执行任务并处理异常，设置最长等待时间，并尝试在超时后取消任务
    :param task_map: 需要执行的任务的map，key是任务的key，val是任务
    :param timeout: 最长等待时间（单位：秒），默认为None，表示无限等待
    :param callback: 所有任务执行完毕后执行的回调函数
    :return: 每个任务执行的结果组成的map，key是任务的key
    """
    executor = ThreadPoolExecutor(max_workers=len(task_map), thread_name_prefix='parallelPool')
    future_to_key = {executor.submit(task): key for key, task in task_map.items()}
    result_map = {}

    # 快速返回
    if timeout is not None and timeout < 0:
        if callback is not None:
            def run_callback():
                try:
                    for future in as_completed(future_to_key):
                        key = future_to_key[future]
                        try:
                            result_map[key] = future.result()
                        except Exception as e:
                            log('parallel_error', {
                                'key': key,
                                'error': e
                            })
                            result_map[key] = None
                finally:
                    executor.shutdown(wait=True)
                    callback(result_map)

            threading.Thread(target=run_callback).start()
        return result_map

    try:
        for future in as_completed(future_to_key, timeout=timeout):
            key = future_to_key[future]
            try:
                result_map[key] = future.result()
            except Exception as e:
                log('parallel_error', {
                    'key': key,
                    'error': e
                })
                result_map[key] = None
    except TimeoutError as e:
        for future in future_to_key:
            future.cancel()
        print(f"Tasks timed out: {e}")
        for key in task_map:
            if key not in result_map:
                result_map[key] = None
    finally:
        executor.shutdown(wait=True)

        # 调用回调函数
        if callback is not None:
            callback(result_map)

    return result_map


def format_msg(format_string: str, *args):
    if not isinstance(format_string, str):
        return str(format_string) if format_string else ''
    try:
        if args:
            args = tuple(desc_obj(arg) for arg in args[0])  # 修改这里，从元组中提取参数
        return format_string.format(*args)
    except:
        return str(format_string)


def desc_obj(obj):
    if obj is None:
        return ''
    elif isinstance(obj, (dict, list, tuple)):
        try:
            ret = json.dumps(obj, ensure_ascii=False, default=str)
            try:
                ret2 = json.loads(ret)
                if isinstance(ret2, str):
                    return ret2
                else:
                    return ret
            except:
                return ret
        except:
            return str(obj)
    elif isinstance(obj, Exception):
        # 在这里实现异常翻译逻辑
        tb = obj.__traceback__
        tb_lines = traceback.format_exception(type(obj), obj, tb)
        return str(obj) + '\n' + ''.join(tb_lines)
    else:
        return str(obj)


def tell_dingding(data: dict = None):
    '''
    https://developers.dingtalk.com/document/robots/customize-robot-security-settings
    :param data: 钉钉消息体
    :return:
    '''
    if not data:
        raise ValueError('消息不可为空')
    # 通知dingding 有钉钉key & 数据量>0
    ding_token = one_runtime.get_config('DINGDING_TOKEN')
    if not ding_token:
        raise ValueError('please set DINGDING_TOKEN')

    if get_ip() != one_runtime.get_config('serverip'):
        log('not tell dingding', {
            'data': data
        })
        return

        # 构造请求URL
    url = "https://oapi.dingtalk.com/robot/send?access_token=" + ding_token
    secret_enc = one_runtime.get_config('DINDDING_SEC')
    # 有密钥
    if secret_enc:
        timestamp = str(round(time.time() * 1000))
        secret = secret_enc
        secret_enc = secret.encode('utf-8')
        string_to_sign = '{}\n{}'.format(timestamp, secret)
        string_to_sign_enc = string_to_sign.encode('utf-8')
        hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
        sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
        url += f'&timestamp={timestamp}&sign={sign}'
    # 设置请求头
    headers = {
        'Content-Type': 'application/json'
    }
    # 发起请求并接收响应
    response = requests.post(url, data=json.dumps(data), headers=headers)
    # 解析响应
    response_data = response.json()
    log('tell_dingding', {
        'response': response_data,
        'data': data
    })
    if response_data.get('errcode') != 0:
        raise ValueError('钉钉消息发送失败，' + response_data.get('errmsg'))


def random_str(length):
    letters_and_digits = string.ascii_letters + string.digits
    return ''.join(random.choice(letters_and_digits) for _ in range(length))


def param_post(url, timeout: int = None, headers: dict[str, str] = None, params: dict = None) -> str:
    """
    发起一个param POST请求。

    :param url: 请求的URL地址
    :param timeout: 超时时间，以秒为单位，如果不提供默认是3s
    :param headers: 请求头字典，如果为空，则不设置自定义请求头
    :param params: POST请求的参数字典，如果为空，则不携带任何参数
    :return: 响应的文本内容，请求失败则返回None
    """
    start_time = time.time()
    try:
        timeout = 3 if not timeout else timeout
        # 将数字键转换为字符串，并处理成一层字典，值为JSON字符串
        flattened_params = {str(key): json.dumps(val) for key, val in params.items()} if params else {}
        response = requests.post(url, timeout=timeout, headers=headers, data=flattened_params)
        response.raise_for_status()  # 如果响应状态码指示错误，则抛出异常
        res = response.text
        log('param_post', {
            'url': url,
            'timeout': timeout,
            'headers': headers,
            'params': params,
            'result': res,
            'cost': (time.time() - start_time) * 1000
        })
        return res
    except Exception as e:
        log('param_post', {
            'url': url,
            'timeout': timeout,
            'headers': headers,
            'params': params,
            'error': e,
            'cost': (time.time() - start_time) * 1000
        })
        raise


def json_post(url, timeout: int = None, headers: dict[str, str] = None, body: dict = None) -> str:
    """
    发起一个json POST请求。

    :param url: 请求的URL地址
    :param timeout: 超时时间，以秒为单位，如果不提供默认是3s
    :param headers: 请求头字典，如果为空，则不设置自定义请求头
    :param body: POST请求的参数字典，如果为空，则不携带任何参数
    :return: 响应的文本内容，请求失败则返回None
    """
    start_time = time.time()
    try:
        timeout = 3 if not timeout else timeout
        # 将数字键转换为字符串，并处理成一层字典，值为JSON字符串
        response = requests.post(url, timeout=timeout, headers=headers, json=body)
        response.raise_for_status()  # 如果响应状态码指示错误，则抛出异常
        res = response.text
        log('json_post', {
            'url': url,
            'timeout': timeout,
            'headers': headers,
            'body': body,
            'result': res,
            'cost': (time.time() - start_time) * 1000
        })
        return res
    except Exception as e:
        log('json_post', {
            'url': url,
            'timeout': timeout,
            'headers': headers,
            'body': body,
            'error': e,
            'cost': (time.time() - start_time) * 1000
        })
        raise


def embedding(text: str = None, model: str = None) -> list[float]:
    if not text or not model:
        raise ValueError('text 和 model 不可为空')
    raise ValueError('待实现')


def call_method(obj, method_string):
    if not obj or not method_string:
        raise ValueError('对象和函数不可为空')
    namespace = {
        'obj': obj
    }
    exec('result = obj.' + method_string, namespace)
    return namespace['result']


def compute_rouge(s1: Union[str, list[str]], s2: Union[str, list[str]]) -> dict:
    '''
    计算文本的rouge
    :param s1:
    :param s2:
    :return:
    '''
    assert isinstance(s1, (list, str)), "s1 should be a string or a list of strings"
    assert isinstance(s2, (list, str)), "s2 should be a string or a list of strings"

    if isinstance(s1, str):
        s1 = [s1]

    if isinstance(s2, str):
        s2 = [s2]

    assert len(s1) == len(s2), "The length of text1 and text2 should be the same"

    score_dict = {
        "rouge-1": [],
        "rouge-2": [],
        "rouge-l": [],
    }

    for t1, t2 in zip(s1, s2):
        hypothesis = list(jieba.cut(t1))
        reference = list(jieba.cut(t2))

        if any([len(hypothesis) <= 0, len(reference) <= 0]):
            log('rouge_skip', {
                'text1': hypothesis,
                'text2': reference
            }, 'skip wrong data')
            continue

        rouge = Rouge()
        scores = rouge.get_scores(' '.join(hypothesis), ' '.join(reference))
        result = scores[0]

        for k, v in result.items():
            score_dict[k].append(round(v["f"], 4))

    for k, v in score_dict.items():
        score_dict[k] = float(np.mean(v))

    return score_dict


def compute_bleu(compares: Union[str, list[str]], stds: Union[str, list[str]]):
    '''
    :param compares: 需要比对的字符串 bleu有方向性注意别传错了
    :param stds: 标准字符串 bleu有方向性注意别传错了
    :return:
    '''
    assert isinstance(compares, (list, str)), "compares should be a string or a list of strings"
    assert isinstance(stds, (list, str)), "stds should be a string or a list of strings"

    if isinstance(compares, str):
        compares = [compares]

    if isinstance(stds, str):
        stds = [stds]

    assert len(compares) == len(stds), "The length of compares and stds should be the same"

    score_list = []
    smoothing = SmoothingFunction().method3
    for t1, t2 in zip(compares, stds):
        hypothesis = list(jieba.cut(t1))
        reference = list(jieba.cut(t2))

        if any([len(hypothesis) <= 0, len(reference) <= 0]):
            log('bleu_skip', {
                'compare': hypothesis,
                'std': reference
            }, 'skip wrong data')
            continue

        score = sentence_bleu([reference], hypothesis, smoothing_function=smoothing)
        score_list.append(round(score, 4))

    return float(np.mean(score_list))


def build_excel(builder=None, excel_name: str = None):
    if builder is None:
        raise ValueError("builder function can not be None")
    data = builder()
    # 将字典转换为JSON字符串
    for item in data:
        for key, value in item.items():
            if isinstance(value, (dict, list)):
                item[key] = json.dumps(value, ensure_ascii=False, default=str)
            elif isinstance(value, bool):
                item[key] = str(value).lower()
            elif isinstance(value, (float, int)):
                item[key] = "" if pd.isna(value) else str(value)
    # 将其转化为DataFrame对象
    df = pd.DataFrame(data)
    for column in df.columns:
        df[column] = df[column].fillna('').astype(str)
    # 获取当前日期并格式化为yyyymmdd
    if not excel_name:
        excel_name = datetime.now().strftime('%Y%m%d%H')
    # 定义文件路径，注意这里的路径可能需要你根据实际情况进行修改
    # macOS系统
    file_path = f'/Users/wangshaopeng/Desktop/{excel_name}.xlsx'
    # 输出到excel
    df.to_excel(file_path, index=False)


def convert_excel(excel_name: str = None, convert=None):
    if excel_name is None or convert is None:
        raise ValueError("excel_name and convert function can not be None")
    file_name = f'/Users/wangshaopeng/Desktop/{excel_name}.xlsx'
    if not os.path.exists(file_name):
        raise FileNotFoundError(f"{file_name} does not exist")

        # 读取Excel，使用第一行作为列名（默认行为）
    df = pd.read_excel(file_name)

    # 转换数据
    converted_data = df.to_dict(orient='records')
    converted_data = list(map(convert, converted_data))

    # 展开item
    expanded_data = []
    for item in converted_data:
        if item is not None:
            if isinstance(item, list):
                expanded_data.extend(item)
            else:
                expanded_data.append(item)
    converted_data = expanded_data
    # 将字典转换为JSON字符串
    for item in converted_data:
        for key, value in item.items():
            if isinstance(value, (dict, list)):
                item[key] = json.dumps(value, ensure_ascii=False, default=str)
            elif isinstance(value, bool):
                item[key] = str(value).lower()
            elif isinstance(value, (float, int)):
                item[key] = "" if pd.isna(value) else str(value)
    # 将转换后的数据保存为新的Excel
    new_df = pd.DataFrame(converted_data)
    new_df.to_excel(file_name.rsplit('.', 1)[0] + "_convert.xlsx", index=False)



def local_to_utc(local_time_str: str, local_tz_str: str) -> str:
    """
    将本地时间转换为 UTC 时间，并格式化为 ISO 8601 格式。

    :param local_time_str: 本地时间字符串（例如 '2023-01-01 00:00:00'）
    :param local_tz_str: 本地时区（例如 'Asia/Shanghai'）
    :return: 转换后的 UTC 时间字符串（ISO 8601 格式）
    """
    # 解析本地时间
    local_time = datetime.strptime(local_time_str, '%Y-%m-%d %H:%M:%S')

    # 获取本地时区
    local_tz = pytz.timezone(local_tz_str)

    # 设置本地时区
    local_time = local_tz.localize(local_time)

    # 转换为 UTC 时间
    utc_time = local_time.astimezone(pytz.utc)

    # 格式化为 ISO 8601 格式
    return utc_time.strftime('%Y-%m-%dT%H:%M:%S')


def search_es(host: str=None, index_name: str=None,match_fields: Dict[str, str] = None,query_string: str=None, start_date: str=None, end_date: str=None, size:int=None) -> List[Dict]:

    """
    查询 Elasticsearch 中的日志，按 query_string 和时间范围过滤
    :param host: Elasticsearch 主机地址（如 'http://localhost:9200' 或 'https://localhost:9200'）
    :param index_name: 要查询的索引名称
    :param query_string: 用于查询的字符串
    :param start_date: 查询的开始时间（ISO 8601 格式）
    :param end_date: 查询的结束时间（ISO 8601 格式）
    :return: 符合条件的日志列表
    """
    # 解析 Elasticsearch 主机地址
    parsed_url = urlparse(host)
    scheme = parsed_url.scheme if parsed_url.scheme else 'http'
    host = parsed_url.hostname if parsed_url.hostname else 'localhost'
    port = parsed_url.port if parsed_url.port else 9200

    # 创建 Elasticsearch 客户端
    es = Elasticsearch([{'scheme': scheme, 'host': host, 'port': port}])

    # 构建查询
    must = [
        {
            "query_string": {
                "query": query_string
            }
        }
    ]

    if start_date and end_date:
        must.append(
            {
                "range": {
                    "timestamp": {  # 替换为你实际的时间字段名
                        "gte": local_to_utc(start_date, 'Asia/Shanghai'),
                        "lte": local_to_utc(end_date, 'Asia/Shanghai'),
                        "format": "strict_date_optional_time"
                    }
                }
            }
        )

    if match_fields:
        for field, value in match_fields.items():
            must.append({
                "match": {
                    field: value
                }
            })

    body = {
        "query": {
            "bool": {
                "must": must
            }
        }
    }
    if size is not None:
        body['size'] = size

    # 执行搜索请求
    response = es.search(index=index_name, body=body)

    # 提取并返回文档部分
    hits = response['hits']['hits']
    return [hit['_source'] for hit in hits]


if __name__ == '__main__':
    print(1)
  #   xx = json_post("https://kkkc.net/v1/chat/completions",30,{"Authorization":'''Bearer sk-DbE7zaQqrZalO5LaE230669e7dF449A8846e47781a50B5Ea'''},{
  #   "model": "gpt-4o",
  #   "messages": [
  #     {
  #       "role": "system",
  #       "content": "You are a helpful assistant."
  #     },
  #     {
  #       "role": "user",
  #       "content": "Hello!"
  #     }
  #   ]
  # })


    # xx = search_es(host="localhost:9200",index_name="logs_index",query_string="TN3kofhBeY92a5u1wBW9RDY5YLoVajTX",size=2)
    # print(xx)

